package com.gemantic.task.config.scheduler;


import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
import java.util.Map;
import javax.annotation.PostConstruct;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.context.annotation.Configuration;

import com.gemantic.task.consumer.TaskConsumer;
import com.gemantic.task.consumer.TaskConsumer2;

import lombok.extern.slf4j.Slf4j;


/**
 * 调度器消费者配置
 *
 * @author xiedi
 * @date 2023/08/29
 */
@Configuration
@AutoConfigureBefore(value = {TaskConsumer.class, TaskConsumer2.class})
@Slf4j
public class SchedulerConsumerConfig {

    private static final String CONSUME_THREAD_MAX = "consumeThreadMax";

    @Value("${task.p1.consume-thread-max:0}")
    private int taskP1ConsumeThreadMax;

    @Value("${task.p2.consume-thread-max:0}")
    private int taskP2ConsumeThreadMax;

    @PostConstruct
    public void init() {
        this.setConsumeThreadMax(TaskConsumer.class, taskP1ConsumeThreadMax);
        this.setConsumeThreadMax(TaskConsumer2.class, taskP2ConsumeThreadMax);
    }


    private void setConsumeThreadMax(Class<? extends RocketMQListener> rClass, int newConsumeThreadMax) {
        try {
            RocketMQMessageListener annotation = rClass.getAnnotation(RocketMQMessageListener.class);
            InvocationHandler invocationHandler = Proxy.getInvocationHandler(annotation);
            Field values = invocationHandler.getClass().getDeclaredField("memberValues");
            values.setAccessible(true);
            @SuppressWarnings("all")
            Map<String, Object> memberValues = (Map<String, Object>) values.get(invocationHandler);
            String topic = (String) memberValues.get("topic");
            int originalValue = (int) memberValues.get(CONSUME_THREAD_MAX);
            if (newConsumeThreadMax > 0 && originalValue != newConsumeThreadMax) {
                memberValues.put(CONSUME_THREAD_MAX, newConsumeThreadMax);
                log.info(String.format("scheduler修改topic[%s]的最大线程数从[%s]调整为[%s]", topic, originalValue,
                        newConsumeThreadMax));
            }
        } catch (Exception ex) {
            log.warn(ExceptionUtils.getStackTrace(ex));
        }
    }


}
